package com.heima.article.listener;

import com.heima.common.util.JsonUtils;
import com.heima.model.article.msg.ArticleBehaviorMsg;
import com.heima.model.article.msg.ArticleBehaviorStreamMsg;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

import static com.heima.common.constants.message.HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC;
import static com.heima.common.constants.message.HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC;

/**
 * 实时流式计算，统计文章分值
 */
@EnableBinding(HotArticleStreamListener.HotArticleStreamProcessor.class)
public class HotArticleStreamListener {

    /**
     * 统计文章分值
     */
    @StreamListener(HOT_ARTICLE_SCORE_STREAM_TOPIC)
    @SendTo(HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
    public KStream<String,String> process(KStream<String,String> input){
        return input.map(new KeyValueMapper<String, String, KeyValue<String,String>>() {
            //value的格式 {"articleId":12312312,"type":1,"num":1}
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //解析value值
                ArticleBehaviorMsg articleBehaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                return new KeyValue<>(articleBehaviorMsg.getArticleId().toString(),value);
            }
        }).groupByKey(Grouped.with(Serdes.String(),Serdes.String())).
                windowedBy(TimeWindows.of(Duration.ofSeconds(15))).
                aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        return null;
                    }
                }, new Aggregator<String, String, String>() {
                    //key:文章id  value：传递的最新需要处理的数据  aggregate：上次聚合结果
                    @Override
                    public String apply(String key, String value, String aggregate) {
                        ArticleBehaviorStreamMsg streamMsg = null;
                        if(StringUtils.isBlank(aggregate)){
                            //上次聚合结果为null，初始化ArticleBehaviorStreamMsg对象
                            streamMsg = new ArticleBehaviorStreamMsg();
                            streamMsg.setArticleId(Long.parseLong(key));
                            streamMsg.setCollection(0);
                            streamMsg.setComment(0);
                            streamMsg.setLikes(0);
                            streamMsg.setViews(0);
                        }else {
                            //解析上传统计结果
                            streamMsg = JsonUtils.toBean(aggregate,ArticleBehaviorStreamMsg.class);
                            //解析value值
                            ArticleBehaviorMsg articleBehaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                            switch (articleBehaviorMsg.getType()){
                                case LIKES:
                                    streamMsg.setLikes(streamMsg.getLikes()+articleBehaviorMsg.getNum());
                                    break;
                                case VIEWS:
                                    streamMsg.setViews(streamMsg.getViews()+articleBehaviorMsg.getNum());
                                    break;
                                case COMMENT:
                                    streamMsg.setComment(streamMsg.getComment()+articleBehaviorMsg.getNum());
                                    break;
                                case COLLECTION:
                                    streamMsg.setCollection(streamMsg.getCollection()+articleBehaviorMsg.getNum());
                                    break;
                            }

                        }
                        return JsonUtils.toString(streamMsg);
                    }
                },Materialized.with(Serdes.String(),Serdes.String())).toStream().
                map(new KeyValueMapper<Windowed<String>, String, KeyValue<String,String>>() {
                    @Override
                    public KeyValue<String,String> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key(),value);
                    }
                });
    }


    public interface HotArticleStreamProcessor{
        @Input(HOT_ARTICLE_SCORE_STREAM_TOPIC)
        KStream<String,String> input();

        @Output(HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
        KStream<String,String> output();
    }
}
